package spark.OrcReadAndWrite.Kryo;

import org.apache.hadoop.hive.ql.io.orc.OrcSerde;
import org.apache.hadoop.hive.ql.io.orc.OrcStruct;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.TaskInputOutputContext;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;


/**
 *
 * 这个本来是工具类，但是在我们这需要在map操作里每次都创建，我们想要优化这个项目，将这个对象的创建放到了外面，但是此时就必须实现序列化，
 * 这个不光是这个类要实现序列化，里面的属性要必须都要实现序列化。
 * 注意:属性实现序列化的要求是不一样的，如果属性没有赋值，或等于null，就不必序列化，如果属性被创建了，那么就必须实现序列化
 * 属性实现序列化的方式是在声明语句最前面加 transient
 *
 *ORC工具类，用来处理orc文件的数据和数据类型
 * @author  王硕
 * @Date 2018年6月10日
 */
//public class ORCUtil implements Serializable{
public class ORCUtil {

	//用来设置ORC文件的TypeInfo
	private StructObjectInspector soi = null;
	
	//用来存储ORC文件的每一行数据
	private OrcStruct orc = null;
	
	private ObjectInspector oi = null;
	
	private List <Object> list = null;
	
	private OrcSerde os = null;
	
	
	/**
	 *根据hive表描述设置读取ORC文件时使用的TypeInfo
	 *<\r>
	 * @param type  ORC格式的hive表描述
	 */
	public void setORCtype(String type){
		TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(type);
		soi = (StructObjectInspector)OrcStruct.createObjectInspector(typeInfo);
	}
	
	/**
	 *ORC格式文件中的数据，相当于hive的一行记录，是由orcNewInputFormat读取的。
	 *这个跟设置读取格式有关系。job的设置
	 *<\r>
	 * @param orcStruct
	 */
	public void setRecord(OrcStruct orcStruct){
		this.orc=orcStruct;
	}
	
	
	public String getData(String key){
		StructField sf = soi.getStructFieldRef(key);
		String value = String.valueOf(soi.getStructFieldData(orc, sf));
		return Utils.isEmpty(value) || value.toLowerCase().equals("null") ? null : value;
		
	}
	
	public void setORCWriteType(String type){
		TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(type);
		oi = TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(typeInfo);
	}
	
	public ORCUtil addAttr(Object... objs){
		if(Utils.isEmpty(list)){
			list = new ArrayList();
		}
		for (Object o : objs) {
			list.add(o);
		}
		return this;
	}
	
	public Writable serialize(){
		if(Utils.isEmpty(os)){
			os = new OrcSerde();
		}
		Writable line = os.serialize(list, oi);
		list = new ArrayList();
		return line;
	}
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public void serialize(TaskInputOutputContext context) throws IOException, InterruptedException {
		if (Utils.isEmpty(os)) {
			os = new OrcSerde();
		}
		Writable line = os.serialize(list, oi);
		list = new ArrayList<Object>();
	    context.write(NullWritable.get(), line);
	}
	
}
